//package com.china.produce.controller;
//
//import com.alibaba.fastjson.JSON;
//import com.google.common.collect.Lists;
//import io.swagger.annotations.Api;
//import io.swagger.annotations.ApiOperation;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.kafka.clients.admin.AdminClient;
//import org.apache.kafka.clients.admin.ListTopicsResult;
//import org.apache.kafka.clients.admin.NewTopic;
//import org.apache.kafka.clients.producer.RecordMetadata;
//import org.springframework.beans.factory.InitializingBean;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.beans.factory.annotation.Qualifier;
//import org.springframework.kafka.core.KafkaTemplate;
//import org.springframework.kafka.support.SendResult;
//import org.springframework.util.concurrent.ListenableFuture;
//import org.springframework.util.concurrent.ListenableFutureCallback;
//import org.springframework.web.bind.annotation.*;
//import java.util.ArrayList;
//import java.util.Collections;
//import java.util.HashMap;
//import java.util.Set;
//import java.util.concurrent.ExecutionException;
//
///**
// * @ClassName
// * @Author niujie
// * @Version
// * @Description
// * @CreateTime 2023/4/23
// */
//@Api(tags = "kafka-")
//@RestController
//@RequestMapping("/kafka")
//@Slf4j
//public class KafkaController {
//
//    private static final String TOPIC_NAME = "test.kafka.ack2";
//
//    private static final String TOPIC_NAME_3 = "test.kafka.ack3";
//
//    @Autowired
//    @Qualifier("kaTemplate")
//    private KafkaTemplate<String, String> kafkaTemplate;
//
//    @Autowired
//    @Qualifier("kafkaAdminClient")
//    private AdminClient adminClient;
//
//    @ApiOperation(value = "发送消息")
//    @PostMapping("/v1/send/{num}")
//    public void sendMessage(@PathVariable("num") String num) {
//        HashMap<String, String> map = new HashMap<>();
//        map.put("test", "lin");
//        for (int i = 0; i < Integer.parseInt(num); i++) {
//            map.put("num", String.valueOf(i));
//            kafkaTemplate.send(TOPIC_NAME, JSON.toJSONString(map));
//        }
//    }
//
//    @ApiOperation(value = "发送消息监听返回结果")
//    @PostMapping("/v1/sendListenResult")
//    public void sendMessageListenResult() {
//        ArrayList<String> message = Lists.newArrayList();
//        message.add("dev");
//        message.add("test");
//        message.add("prod");
//        //发送消息到指定topic
//        ListenableFuture<SendResult<String, String>> sendRes = kafkaTemplate.send(TOPIC_NAME_3, JSON.toJSONString(message));
//        sendRes.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
//            @Override
//            public void onFailure(Throwable ex) {
//                log.info("发送失败: {}", ex.getMessage());
//            }
//
//            @Override
//            public void onSuccess(SendResult<String, String> result) {
//                RecordMetadata recordMetadata = result.getRecordMetadata();
//                log.info("success topic: {}  partition: {} offset: {}", recordMetadata.topic(), recordMetadata.partition(), recordMetadata.offset());
//            }
//        });
//    }
//
//    @ApiOperation(value = "获取所有的topics")
//    @GetMapping("/v1/getAllKafkaTopics")
//    public void getAllKafkaTopics() throws ExecutionException, InterruptedException {
//
//        ListTopicsResult listTopics = adminClient.listTopics();
//
//        Set<String> topics = listTopics.names().get();
//
//        topics.forEach(s-> System.out.println(s));
//    }
//
//    @ApiOperation(value = "删除指定topic")
//    @PostMapping("/v1/delKafkaTopic/{name}")
//    public void delKafkaTopic(@PathVariable("name") String name) {
//
//        adminClient.deleteTopics(Collections.singletonList(name));
//    }
//
//    @ApiOperation(value = "创建指定topic")
//    @PostMapping("/v1/createKafkaTopic/{name}")
//    public void createKafkaTopic(@PathVariable("name") String name) throws InterruptedException {
//
//        NewTopic topic = new NewTopic(name, 2, (short) 1);
//
//        adminClient.createTopics(Collections.singletonList(topic));
//
//        Thread.sleep(1000);
//    }
//}
